package com.we.flink.userlabelplatform.tag;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.flink.userlabelplatform.table.TableUserLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;

public class TagUserLevel {
    public long uid;
    public String user_level;
    public Integer account_stauts;
    public Integer amount_retrial_output_time;
    public Float loan_actual_line;
    public Float loan_line;
    /**
     * 3: ＞IRR36%
     *
     * <p>2:（IRR24%，IRR36]
     *
     * <p>1: ≤IRR24%
     *
     * <p>0: 未知
     */
    public Integer irr_year_rate_group;
    /** 非标签,仅用来比较判断是否更新 */
    public int updated_at;

    public static Logger LOG = LoggerFactory.getLogger(TagUserLevel.class);

    public TagUserLevel() {}

    public TagUserLevel(
            long uid,
            String user_level,
            Integer account_stauts,
            Integer amount_retrial_output_time,
            Float loan_actual_line,
            Float loan_line,
            Integer irr_year_rate_group,
            int updated_at) {
        this.uid = uid;
        this.user_level = user_level;
        this.account_stauts = account_stauts;
        this.amount_retrial_output_time = amount_retrial_output_time;
        this.loan_actual_line = loan_actual_line;
        this.loan_line = loan_line;
        this.irr_year_rate_group = irr_year_rate_group;
        this.updated_at = updated_at;
    }

    public static TagUserLevel init(long uid) {
        return new TagUserLevel(uid, "", -1, 0, new Float(-1), new Float(-1), 0, 0);
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableUserLevel.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUserLevel>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUserLevel.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableUserLevel>>() {
                                    @Override
                                    public boolean filter(Tuple2<Boolean, TableUserLevel> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TagUserLevel-filter")
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableUserLevel>, Long>() {
                                    @Override
                                    public Long getKey(Tuple2<Boolean, TableUserLevel> value)
                                            throws Exception {
                                        return value.f1.getUid();
                                    }
                                })
                        .flatMap(
                                new RichFlatMapFunction<Tuple2<Boolean, TableUserLevel>, String>() {
                                    ValueState<TagUserLevel> tableUserLevelVState;

                                    @Override
                                    public void open(Configuration parameters) throws Exception {
                                        tableUserLevelVState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<
                                                                        TagUserLevel>(
                                                                        "UserLevelVState",
                                                                        TagUserLevel.class));
                                    }

                                    @Override
                                    public void flatMap(
                                            Tuple2<Boolean, TableUserLevel> value,
                                            Collector<String> out)
                                            throws Exception {
                                        try {
                                            long uid = value.f1.getUid();
                                            int updated_at = value.f1.getUpdated_at();
                                            int account_status = value.f1.getAccount_status();
                                            String periods_money_scope =
                                                    value.f1.getPeriods_money_scope();
                                            String level = value.f1.getLevel();
                                            BigDecimal loan_amount = value.f1.getLoan_amount();
                                            BigDecimal loan_actual_amount =
                                                    value.f1.getLoan_actual_amount();
                                            int tmpIrr_year_rate_group = 0;
                                            JSONObject jsonObject = new JSONObject();
                                            jsonObject.put("uid", uid);

                                            TagUserLevel tmpState = tableUserLevelVState.value();
                                            TagUserLevel curState =
                                                    tmpState == null
                                                            ? TagUserLevel.init(uid)
                                                            : tmpState;

                                            /** 取最新时间的各个指标 */
                                            if (updated_at > curState.getUpdated_at()) {

                                                Integer maxPeriod = 0;
                                                Double rate = Double.valueOf(0);

                                                JSONArray objects =
                                                        JSONArray.parseArray(periods_money_scope);
                                                for (int i = 0; i < objects.size(); i++) {
                                                    JSONObject json = objects.getJSONObject(i);
                                                    Integer period = json.getInteger("period");
                                                    if (period != null && maxPeriod < period) {
                                                        maxPeriod = Math.min(period, 12);
                                                        rate = json.getDouble("rate");
                                                    }
                                                }

                                                /** rate 为空 --> 月综费率未知 */
                                                if (maxPeriod != null && rate != null) {
                                                    switch (maxPeriod) {
                                                        case 3:
                                                            if (rate <= 0.0134) {
                                                                tmpIrr_year_rate_group = 1;
                                                            } else if (rate > 0.0134
                                                                    && rate <= 0.02) {
                                                                tmpIrr_year_rate_group = 2;
                                                            } else {
                                                                tmpIrr_year_rate_group = 3;
                                                            }
                                                            break;
                                                        case 6:
                                                            if (rate <= 0.0118) {
                                                                tmpIrr_year_rate_group = 1;
                                                            } else if (rate > 0.0118
                                                                    && rate <= 0.0179) {
                                                                tmpIrr_year_rate_group = 2;
                                                            } else {
                                                                tmpIrr_year_rate_group = 3;
                                                            }
                                                            break;
                                                        case 9:
                                                            if (rate <= 0.0114) {
                                                                tmpIrr_year_rate_group = 1;
                                                            } else if (rate > 0.0114
                                                                    && rate <= 0.0173) {
                                                                tmpIrr_year_rate_group = 2;
                                                            } else {
                                                                tmpIrr_year_rate_group = 3;
                                                            }
                                                            break;
                                                        case 12:
                                                            if (rate <= 0.0112) {
                                                                tmpIrr_year_rate_group = 1;
                                                            } else if (rate > 0.0112
                                                                    && rate <= 0.0171) {
                                                                tmpIrr_year_rate_group = 2;
                                                            } else {
                                                                tmpIrr_year_rate_group = 3;
                                                            }
                                                            break;
                                                        default:
                                                            /** 期数不在3，6，9，12 */
                                                            tmpIrr_year_rate_group = 0;
                                                            LOG.warn(
                                                                    "uid: "
                                                                            + uid
                                                                            + " maxPeriod input error: "
                                                                            + maxPeriod);
                                                            break;
                                                    }
                                                } else {
                                                    tmpIrr_year_rate_group = 0;
                                                }

                                                if (!curState.account_stauts.equals(account_status)
                                                        || !curState.user_level.equals(level)
                                                        || !curState.loan_actual_line.equals(
                                                                loan_actual_amount.floatValue())
                                                        || !curState.loan_line.equals(
                                                                loan_amount.floatValue())
                                                        || !curState.irr_year_rate_group.equals(
                                                                tmpIrr_year_rate_group)) {

                                                    curState.account_stauts = account_status;
                                                    curState.user_level = level;
                                                    curState.updated_at = updated_at;
                                                    curState.loan_line = loan_amount.floatValue();
                                                    curState.loan_actual_line =
                                                            loan_actual_amount.floatValue();
                                                    curState.irr_year_rate_group =
                                                            tmpIrr_year_rate_group;

                                                    /** 更新才往sink kafka写数据 */
                                                    jsonObject.put(
                                                            "account_stauts",
                                                            (curState.account_stauts == -1
                                                                    ? null
                                                                    : curState.account_stauts));
                                                    jsonObject.put(
                                                            "loan_actual_line",
                                                            (curState.loan_actual_line == -1
                                                                    ? null
                                                                    : curState.loan_actual_line));
                                                    jsonObject.put(
                                                            "loan_line",
                                                            (curState.loan_line == -1
                                                                    ? null
                                                                    : curState.loan_line));
                                                    jsonObject.put(
                                                            "user_level", curState.user_level);
                                                    jsonObject.put(
                                                            "irr_year_rate_group",
                                                            curState.irr_year_rate_group);
                                                }
                                            }
                                            /** 最近一次贷中额度重审输出时间 */
                                            if (account_status == 8
                                                    && curState.amount_retrial_output_time
                                                            < updated_at) {
                                                curState.amount_retrial_output_time = updated_at;
                                                jsonObject.put(
                                                        "amount_retrial_output_time",
                                                        (curState.amount_retrial_output_time == 0
                                                                ? null
                                                                : curState.amount_retrial_output_time));
                                            }

                                            if (jsonObject.size() > 1) {
                                                out.collect(jsonObject.toString());
                                                tableUserLevelVState.update(curState);
                                            }

                                        } catch (Exception E) {
                                            LOG.error("flatMap: " + E.getMessage());
                                            LOG.error("flatmap input.f1: " + value.f1);
                                        }
                                    }
                                });

        return out;
    }

    public long getUid() {
        return uid;
    }

    public String getUser_level() {
        return user_level;
    }

    public Integer getAccount_stauts() {
        return account_stauts;
    }

    public int getUpdated_at() {
        return updated_at;
    }

    public Float getLoan_actual_line() {
        return loan_actual_line;
    }

    public Float getLoan_line() {
        return loan_line;
    }

    @Override
    public String toString() {
        return "UserLevelTag{"
                + "uid="
                + uid
                + ", user_level='"
                + user_level
                + '\''
                + ", account_stauts="
                + account_stauts
                + ", loan_actual_line="
                + loan_actual_line
                + ", loan_line="
                + loan_line
                + ", updated_at="
                + updated_at
                + '}';
    }
}
